package main

//
//import (
//	"encoding/binary"
//	"fmt"
//	"io"
//	"net"
//	"rpc/msg"
//	"strconv"
//	"sync"
//	"time"
//)
//
//func main() {
//
//	var wg sync.WaitGroup
//
//	client := NewClient(500, "127.0.0.1:8080")
//
//	now := time.Now()
//	for i := 0; i < 1000000; i++ {
//		wg.Add(1)
//		go func() {
//			_, _ = client.Send([]byte("hello, go" + strconv.Itoa(i)))
//			//fmt.Println(string(v))
//			wg.Done()
//		}()
//	}
//
//	wg.Wait()
//	fmt.Println("elapse: ", time.Since(now))
//
//}
//
//const (
//	BatchSize = 64 * 1024 * 1024
//)
//
//type Client struct {
//	writeLock chan struct{}
//	sendMerge chan *msg.Msg
//
//	connection []*Conn
//
//	ack *sync.Map
//}
//
//type Conn struct {
//	con        net.Conn
//	lock       *sync.Mutex
//	state      int
//	sendPacket chan *[]byte
//}
//
//type Packet struct {
//	header uint64
//	data   []byte
//}
//
//func NewClient(connPool int, addr string) *Client {
//	cli := Client{
//		writeLock:  make(chan struct{}, 1),
//		sendMerge:  make(chan *msg.Msg, 0),
//		connection: make([]*Conn, 0),
//		ack:        new(sync.Map),
//	}
//
//	for i := 0; i < connPool; i++ {
//		conn, err := net.Dial("tcp4", addr)
//		if err != nil {
//			panic(err)
//		}
//		cli.connection = append(cli.connection, &Conn{
//			con:        conn,
//			lock:       new(sync.Mutex),
//			state:      0,
//			sendPacket: make(chan *[]byte, 10240),
//		})
//	}
//
//	cli.doRead()
//	cli.doSend2()
//
//	return &cli
//}
//
//func (client *Client) Send(data []byte) ([]byte, error) {
//
//	message := msg.NewMsg(data)
//	resultChan := make(chan *msg.Msg)
//	client.ack.Store(message.SeqNo, resultChan)
//	defer func() {
//		client.ack.Delete(message.SeqNo)
//	}()
//
//	select {
//	case client.sendMerge <- message:
//		// 请求被合并了，等待结果
//		rep := <-resultChan
//		return rep.Data, nil
//
//	case client.writeLock <- struct{}{}:
//
//	}
//
//	client.SendWithLock(message)
//
//	rep := <-resultChan
//	return rep.Data, nil
//}
//
//func (client *Client) SendWithLock(request *msg.Msg) {
//	var batch = []*msg.Msg{request}
//	mergeLimit := uint64(BatchSize)
//merge:
//	for mergeLimit > 0 {
//		select {
//		case msg := <-client.sendMerge:
//			if msg.Data != nil {
//				if msg.MsgTotalLength() > mergeLimit {
//					break merge
//				}
//				batch = append(batch, msg)
//				mergeLimit = mergeLimit - msg.MsgTotalLength()
//			}
//		default:
//			break merge
//		}
//	}
//	<-client.writeLock
//	client.asyncSend(batch)
//}
//
//func (client *Client) asyncSend(batch []*msg.Msg) {
//	i := time.Now().UnixMicro() % int64(len(client.connection))
//	sendPacket := client.connection[i].sendPacket
//	if len(batch) > 10 {
//		fmt.Println(len(batch))
//	}
//	var packetSize uint64
//	for _, req := range batch {
//		packetSize = packetSize + req.Length + 8
//	}
//	packet := make([]byte, 0, packetSize)
//	for _, req := range batch {
//		packet = append(packet, req.Encode()...)
//	}
//	sendPacket <- &packet
//}
//
//func (client *Client) doRead() {
//	for _, conn := range client.connection {
//		go func() {
//			for {
//				header := make([]byte, 16)
//				_, err := conn.con.Read(header)
//				if err != nil {
//					if err == io.EOF {
//						return
//					}
//					return
//				}
//
//				l := binary.BigEndian.Uint64(header[:8])
//				seqNo := binary.BigEndian.Uint64(header[8:16])
//
//				data := make([]byte, l-8)
//				_, err = conn.con.Read(data)
//				if err != nil {
//					if err == io.EOF {
//						return
//					}
//					return
//				}
//				rep := &msg.Msg{
//					Length: l,
//					SeqNo:  seqNo,
//					Data:   data,
//				}
//
//				value, _ := client.ack.Load(seqNo)
//				repChan, ok := value.(chan *msg.Msg)
//				if !ok {
//					fmt.Println("+++++", seqNo, *rep)
//				}
//				repChan <- rep
//			}
//		}()
//	}
//}
//
//func (client *Client) doSend(batch []*msg.Msg) {
//
//	i := time.Now().UnixNano() % int64(len(client.connection))
//
//	conn := client.connection[i]
//
//	packet := make([]byte, 0)
//	for _, req := range batch {
//		packet = append(packet, req.Encode()...)
//	}
//	conn.lock.Lock()
//	defer conn.lock.Unlock()
//	_, _ = conn.con.Write(packet)
//}
//
//func (client *Client) doSend2() {
//	for _, conn := range client.connection {
//		go func() {
//			for {
//				select {
//				case packet := <-conn.sendPacket:
//					conn.lock.Lock()
//					conn.con.Write(*packet)
//					conn.lock.Unlock()
//				}
//			}
//		}()
//	}
//
//}
